package com.fy.service.service.impl;

import com.fy.common.constant.Constants;
import com.fy.common.model.im.MessageModel;
import com.fy.service.service.MessageService;
import com.xiaoju.uemc.tinyid.client.utils.TinyId;
import lombok.SneakyThrows;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.shaded.org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.hbase.shaded.org.apache.commons.httpclient.util.DateUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.unit.DataUnit;

import java.io.IOException;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.Map;


@Service
public class MessageServiceImpl implements MessageService {

    /**
     * 内部已实现线程安全的连接池
     */

    @Autowired
    private Connection hbaseConnection;


    /**
     * 持久化聊天消息
     *
     * @param messageModel 一对一消息持久化对象
     * @throws IOException
     */

    @SneakyThrows
    @Override
    public void  saveSingleMessage(MessageModel messageModel) {
        //设置时间戳
        Date date = new Date();
        String datastr = DateUtil.formatDate(date,"yyyy-MM-dd hh:MM:ss");

        messageModel.setDateStr(datastr);
        messageModel.setMessageId(TinyId.nextId(Constants.TINYID_MESSAGE_ID));
        messageModel.setSendTime(date.getTime());
        Table table = hbaseConnection.getTable(TableName.valueOf("message:message"));

        String messagePrefix = "single_";


        //配置一条数据
        //前缀的设计为用户id_消息id  这样可通过前缀过滤来查询数据 ，也可使同一个用户的数据分配到同一个region 大大提高性能 拆分策略请看DelimitedKeyPrefixRegionSplitPolicy策略
        //如果是group消息则 策略为 groupId_messageId
        //存储发送人消息
        Put fromMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getFrom() + "_" + messageModel.getTo() + "_" + date.getTime() + "_" + messageModel.getMessageId()));
        //存储接收人消息
        Put toMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getTo() + "_" + messageModel.getFrom() + "_" + date.getTime() + "_" + messageModel.getMessageId()));
        Map<String, Object> messageMap = BeanUtils.describe(messageModel);
        messageMap.remove("token");
        String columnFamily = "message";
        for (String column : messageMap.keySet()) {
            if (messageMap.get(column) != null) {
                fromMessage.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(messageMap.get(column).toString()));
                toMessage.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(messageMap.get(column).toString()));
            }
        }
        fromMessage.setTimestamp(date.getTime());
        toMessage.setTimestamp(date.getTime());


        ArrayList<Put> puts = new ArrayList<>();
        puts.add(fromMessage);
        puts.add(toMessage);
        table.put(puts);

    }


    /**
     * 持久化群消息
     */

    @SneakyThrows
    @Override
    public void saveGroupMessage(MessageModel messageModel) {
        //设置时间戳
        Date date = new Date();
        String datastr = DateUtil.formatDate(date,"yyyy-MM-dd hh:MM:ss");

        messageModel.setDateStr(datastr);

        messageModel.setMessageId(TinyId.nextId(Constants.TINYID_MESSAGE_ID));
        messageModel.setSendTime(date.getTime());
        Table table = hbaseConnection.getTable(TableName.valueOf("message:message"));
        String messagePrefix = "group_";
        //配置一条数据
        //前缀的设计为用户id_消息id  这样可通过前缀过滤来查询数据 ，也可使同一个用户的数据分配到同一个region 大大提高性能 拆分策略请看DelimitedKeyPrefixRegionSplitPolicy策略
        //如果是group消息则 策略为 groupId_messageId

        Put groupMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getFrom() + "_" + messageModel.getToGroup() + "_" + System.currentTimeMillis() + "_" + messageModel.getMessageId()));
        //存储接收人消息
        //Put toMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getTo()+ "_" + messageModel.getFrom() + "_" + messageModel.getMessageId()));
        Map<String, Object> messageMap = BeanUtils.describe(messageModel);
        messageMap.remove("token");
        String columnFamily = "message";
        for (String column : messageMap.keySet()) {
            if (messageMap.get(column) != null) {
                groupMessage.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(messageMap.get(column).toString()));

            }
        }
        groupMessage.setTimestamp(date.getTime());
        ArrayList<Put> puts = new ArrayList<>();
        puts.add(groupMessage);
        table.put(puts);
    }

    @Override
    @SneakyThrows
    public void saveFromMessage(MessageModel messageModel) {
        //设置时间戳
        Date date = new Date();
        String datastr = DateUtil.formatDate(date,"yyyy-MM-dd hh:MM:ss");
        messageModel.setDateStr(datastr);
        messageModel.setMessageId(TinyId.nextId(Constants.TINYID_MESSAGE_ID));
        messageModel.setSendTime(date.getTime());
        Table table = hbaseConnection.getTable(TableName.valueOf("message:message"));
        String messagePrefix = "single_";
        //存储接收人消息
        Put toMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getFrom() + "_" + messageModel.getTo() + "_" + date.getTime() + "_" + messageModel.getMessageId()));
        Map<String, Object> messageMap = BeanUtils.describe(messageModel);
        messageMap.remove("token");
        String columnFamily = "message";
        for (String column : messageMap.keySet()) {
            if (messageMap.get(column) != null) {
                toMessage.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(messageMap.get(column).toString()));
            }
        }
        toMessage.setTimestamp(date.getTime());
        ArrayList<Put> puts = new ArrayList<>();
        puts.add(toMessage);
        table.put(puts);
    }

    @Override
    @SneakyThrows
    public void saveToMessage(MessageModel messageModel) {
        //设置时间戳
        Date date = new Date();
        String datastr = DateUtil.formatDate(date,"yyyy-MM-dd hh:MM:ss");
        messageModel.setDateStr(datastr);
        messageModel.setMessageId(TinyId.nextId(Constants.TINYID_MESSAGE_ID));
        messageModel.setSendTime(date.getTime());
        Table table = hbaseConnection.getTable(TableName.valueOf("message:message"));
        String messagePrefix = "single_";
        //存储接收人消息
        Put toMessage = new Put(Bytes.toBytes(messageModel.getTenantId() + "_" + messagePrefix + messageModel.getTo() + "_" + messageModel.getFrom() + "_" + date.getTime() + "_" + messageModel.getMessageId()));
        Map<String, Object> messageMap = BeanUtils.describe(messageModel);
        messageMap.remove("token");
        String columnFamily = "message";
        for (String column : messageMap.keySet()) {
            if (messageMap.get(column) != null) {
                toMessage.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(messageMap.get(column).toString()));
            }
        }
        toMessage.setTimestamp(date.getTime());
        ArrayList<Put> puts = new ArrayList<>();
        puts.add(toMessage);
        table.put(puts);
    }
}
